/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.agent.service;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import cn.ac.iie.agent.tools.PatternTools;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

/**
 *
 * @author zhangyun
 */
public class HdfsUploadService {

    private static Logger logger = Logger.getLogger(HdfsUploadService.class);
    long complete = 0;//完成上传的文件数
    public String status = "";
    public String desc = "";

    public void upload(String localPath, String fileType, String subjobid, String hdfsPath, String name, String timeout, Map<String, String> rsMap) {
        File fileRoot = new File(localPath);
        if (!fileRoot.isDirectory()) {
            logger.info(subjobid + " localPath is not dir");
            rsMap.put("status", "fail");
            rsMap.put("desc", "localPath is not a dir");
            return;
        }
        String[] allFileList = fileRoot.list();
        //match file
        List<String> fileList = new ArrayList<String>();
        List<String> okFileList = new ArrayList<String>();
        List<String> finalFileList = new ArrayList<String>();
        for (int i = 0; i < allFileList.length; i++) {
            if (allFileList[i].endsWith(".ok")) {
                okFileList.add(allFileList[i]);
            } else {
                fileList.add(allFileList[i]);
            }
        }
        logger.info(subjobid + " file total number: " + fileList.size());
        for (int i = 0; i < fileList.size(); i++) {
            boolean flag = true;
            flag = PatternTools.matchStr(fileType, allFileList[i]);
            if (!flag) {
                continue;
            }
            for (int j = 0; j < okFileList.size(); j++) {
                if ((fileList.get(i) + ".ok").equals(okFileList.get(j))) {
                    finalFileList.add(fileList.get(i));
                }
            }
        }
        Integer totalSize = finalFileList.size();
        rsMap.put("total", totalSize.toString());//匹配上的文件总数，请注意是一半，没算.ok
        logger.info(subjobid + " file upload total number: " + finalFileList.size());
        for (int i = 0; i < finalFileList.size(); i++) {
            logger.info(subjobid + " file[" + i + "]: " + finalFileList.get(i));
        }

        List<String> finalList = new ArrayList<String>();
        if (finalFileList.size() == 0) {//没有要采集的文件
            status = "sucess";
            desc = "have no files for uploading";
            rsMap.put("status", "sucess");
            rsMap.put("desc", "have no files for uploading");
            logger.info(subjobid + " no files for collecting");
        }
        for (int i = 0; i < finalFileList.size(); i++) {
            finalList.add(finalFileList.get(i));
            if (finalList.size() >= 1000) {
                logger.info(subjobid + " >=1000");
                uploadPoolTask(finalList, name, localPath, hdfsPath, subjobid, timeout);
                finalList.clear();
            }
            if (i == finalFileList.size() - 1 && !finalList.isEmpty()) {
                uploadPoolTask(finalList, name, localPath, hdfsPath, subjobid, timeout);
            }
        }
        Long completeInt = complete;
        rsMap.put("complete", completeInt.toString());

        //分析状态信息
        logger.info(subjobid + " status:" + status);
        if (status.contains("fail") && status.contains("sucess")) {
            status = "half_sucess";
            rsMap.put("status", status);
            rsMap.put("desc", desc);
        }
        if (status.contains("fail") && !status.contains("sucess")) {
            status = "fail";
            rsMap.put("status", status);
            rsMap.put("desc", desc);
        }
        if (!status.contains("fail") && status.contains("sucess")) {
            status = "sucess";
            rsMap.put("status", status);
            rsMap.put("desc", "sucess");
        }
    }

    public void uploadPoolTask(List<String> finalFileList, String name, String localPath, String hdfsPath, String subjobid, String timeout) {

        //线程池
        long begin = System.currentTimeMillis();
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(finalFileList.size()),
                new ThreadPoolExecutor.DiscardOldestPolicy());
        for (int i = 0; i < finalFileList.size(); i++) {
            threadPool.execute(new LocalThreadPoolTask(name, localPath + "/" + finalFileList.get(i), hdfsPath, subjobid, this));
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ex) {
            logger.error(subjobid + " sleep error ", ex);
        }
        while (true) {//直至所有线程结束，才继续
            int activeThread = threadPool.getActiveCount();
            if (activeThread == 0) {
                break;
            } else {
                try {
                    long end = System.currentTimeMillis();
                    long middle = (end - begin) / 1000;
                    long tmpTimeout = Long.parseLong(timeout);
                    if (middle < tmpTimeout) {
                        logger.info(subjobid + " runtime:" + middle + " timeout:" + timeout);
                        Thread.sleep(3000);
                    } else {
                        logger.info(subjobid + " upload is killed, because of timeout!");
                        changeStatus("fail");
                        changeDesc("timeout");
                        threadPool.shutdown();
                        return;
                    }
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    logger.error(subjobid + " sleep error ", ex);
                }
            }
        }
        threadPool.shutdown();
    }

//统计上传成功文件个数
    public void plusSucessUploadFileNumber() {
        complete = complete + 1;
    }

    //修改上传状态
    public void changeStatus(String st) {
        status = status + "|" + st;
    }

    //修改上传描述
    public void changeDesc(String pDesc) {
        desc = desc + "|" + pDesc;
    }
}
